package manager;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.Reader;
import java.util.ArrayList;
import java.util.List;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Base64;

import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.PropertiesCredentials;
import com.amazonaws.regions.Region;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.ec2.AmazonEC2Client;
import com.amazonaws.services.ec2.model.Instance;
import com.amazonaws.services.s3.AmazonS3Client;
import com.amazonaws.services.s3.model.S3Object;
import com.amazonaws.services.s3.model.S3ObjectInputStream;
import com.amazonaws.services.sqs.AmazonSQSClient;
import com.amazonaws.services.sqs.model.DeleteMessageRequest;
import com.amazonaws.services.sqs.model.Message;
import com.amazonaws.services.sqs.model.ReceiveMessageRequest;
import com.amazonaws.services.sqs.model.SendMessageRequest;
import com.amazonaws.util.StringInputStream;

import utils.AmazonConstants;
import utils.AmazonUtils;

public class Manager {


	public static void main(String[] args) {
		//args = imagesPerWorker  imagesS3URL
		
		int imagesPerWorker = -1;
		String imageListURLKey = null;
		AmazonS3Client s3 = null;
		AmazonEC2Client ec2 = null;
		AmazonSQSClient sqs = null;
		String jarS3URL = null;
		String m2lqURL = null;
		String l2mqURL = null;
		String m2wqURL = null;
		String w2mqURL = null;
		List<Instance> workerInstances;
		
		try{
			StringInputStream sip = new StringInputStream(AmazonConstants.AMAZON_CARDENTIALS);
			AWSCredentials credentials = new PropertiesCredentials(sip);
			s3 = new AmazonS3Client(credentials);
			ec2 = new AmazonEC2Client(credentials);
			sqs = new AmazonSQSClient(credentials);
			Region usWest2 = Region.getRegion(Regions.US_WEST_2);
			sqs.setRegion(usWest2);
			s3.setRegion(usWest2);
			
			
			//creating queueS
			m2lqURL = AmazonUtils.createQueue(sqs, AmazonConstants.MANAGER_2_LOCAL_QUEUE_NAME);
			l2mqURL = AmazonUtils.createQueue(sqs, AmazonConstants.LOCAL_2_MANAGER_QUEUE_NAME);
			m2wqURL = AmazonUtils.createQueue(sqs, AmazonConstants.MANAGER_2_WORKER_QUEUE_NAME);
			w2mqURL = AmazonUtils.createQueue(sqs, AmazonConstants.WORKER_2_MANAGER_QUEUE_NAME);
			
//			if (args.length < 2){
//				error(m2lqURL,"manager error : usege: <imagesPerWorker> <imageFilePath> ",sqs);
//				System.exit(-1);
//			}

			ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest(l2mqURL);
			receiveMessageRequest.setMaxNumberOfMessages(1);
			List<Message> messages = sqs.receiveMessage(receiveMessageRequest).getMessages();
			if (!messages.isEmpty()){
				Message message = messages.get(0);
				//delete message
				String messageRecieptHandle = messages.get(0).getReceiptHandle();
				sqs.deleteMessage(new DeleteMessageRequest(l2mqURL, messageRecieptHandle));
				String body = message.getBody();
				if (body.startsWith("new task ")){
					String[] split = body.split(" ");
					try{ 
						imagesPerWorker = Integer.parseInt(split[2]);
						imageListURLKey = split[3];
					}catch (NumberFormatException e) {
						error(m2lqURL,"Manager error : there was a number format exception tring to parse \"" + split[2] + "\"" + AmazonConstants.LINE_SEPERATOR  + e.getMessage(),sqs);
						System.exit(-1);
					}
				}
			}
			
			
			
			String[] images = readImagListFile(s3, imageListURLKey);
			int numWorkersInstances = (int) Math.floor(images.length/imagesPerWorker); 
			
			for (String imageUrl : images) {
				sqs.sendMessage(new SendMessageRequest(m2wqURL, "new image task " + imageUrl));	
			}
			//creating workers instances
			workerInstances = AmazonUtils.createInstances(numWorkersInstances, numWorkersInstances, getWorkerScript(jarS3URL,imagesPerWorker), ec2, "worker");
			

			receiveMessageRequest = new ReceiveMessageRequest(w2mqURL);
			receiveMessageRequest.setMaxNumberOfMessages(1);
			int msgCounter = 0;
			

			
			boolean shouldStop = false;
			Thread.sleep(3000);
			StringBuilder htmlSB = new StringBuilder("<html><body>");
			while(!shouldStop){
				messages = sqs.receiveMessage(receiveMessageRequest).getMessages();
				if (!messages.isEmpty()){
					Message message = messages.get(0);
					//delete message
					String messageRecieptHandle = messages.get(0).getReceiptHandle();
					sqs.deleteMessage(new DeleteMessageRequest(w2mqURL, messageRecieptHandle));
					
					msgCounter++;
					System.out.println("  Message");
					System.out.println("    MessageId:     " + message.getMessageId());
					System.out.println("    Body:          " + message.getBody());
					for (Entry<String, String> entry : message.getAttributes().entrySet()) 
					{
						System.out.println("  Attribute");
						System.out.println("    Name:  " + entry.getKey());
						System.out.println("    Value: " + entry.getValue());
					}
					
					String body = message.getBody();
					if (body.startsWith("done image task ")){
					htmlSB.append(body.substring("done image task ".length()-1));
					}
					
				}else{
					Thread.sleep(3600);
				}
				if (msgCounter < images.length){
					
				}else{
					shouldStop = true;
					
				}
			}
			htmlSB.append("</body></html>");
			File file = new File("result.html");
			OutputStream fos = new FileOutputStream(file);
			fos.write(htmlSB.toString().getBytes());
			fos.flush();
			String resultFileUrl = AmazonUtils.uploadFileToS3(file, s3, AmazonConstants.BUCKET_NAME, "result.html");
			
			sqs.sendMessage(new SendMessageRequest(m2lqURL, "done task result.html"));
			
		}catch (Exception e) {
			error(m2lqURL,"there was an error :" + e.getMessage(),sqs);
			
			e.printStackTrace();
		}
		
	}

	public static String[] readImagListFile(AmazonS3Client s3, String imageListURLKey)
			throws IOException {
		S3Object imageListObj = AmazonUtils.readObjectFromS3(s3, AmazonConstants.BUCKET_NAME, imageListURLKey);
		S3ObjectInputStream imageListStream = imageListObj.getObjectContent();
		Reader r = new InputStreamReader(imageListStream);

		StringBuilder sb = new StringBuilder();
		char[] chars = new char[4*1024];
		int len;
		while((len = r.read(chars))>=0) {
		    sb.append(chars, 0, len);
		}
		
		imageListStream.abort();
		imageListStream.close();
		
		String[] l = sb.toString().split("\n");
		return l;
	}

	private static String getWorkerScript(String jarS3URL, int imagePerWorker) {
		ArrayList<String> lines = new ArrayList<String>();
		lines.add("#! /bin/bash");
		if (jarS3URL==null){
			jarS3URL = "https://s3-us-west-2.amazonaws.com/ass1bucket/ass1.jar";
		}
		lines.add("wget "  + jarS3URL + " -O ass1.jar --no-check-certificate");
		lines.add("java -cp ass1.jar worker.Worker");
		lines.add("shutdown -h 0");
		String str = new String(Base64.encodeBase64(AmazonUtils.joinStrings(lines, "\n").getBytes()));
		return str;
	}

	private static void error(String qUrl, String msg, AmazonSQSClient sqs) {
		if (sqs != null){
			sqs.sendMessage(new SendMessageRequest(qUrl,"manager error : " + msg));
		}else{
			System.err.println(msg);
		}
	}
}
